package com.atguigu.state;

import com.atguigu.pojo.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
/*
* 按键分区状态 - 聚合状态 - AggregatingState
*
* AggregatingState：单值状态，不过有聚合功能，会按照定义好的聚合逻辑对存入到状态中的数据进行聚合处理
*
*     OUT get():  获取状态值
*     void add (IN value): 将给定的数据添加到状态中
*     void clear():清空状态
*
* */

public class Flink09_KeyedAggregatingState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //s1,100,1000
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));
                        }
                );
        ds.print("INPUT");

        //todo 需求：计算每种传感器的平均水平(怎么求平均值呢？)
        ds
                .keyBy(WaterSensor::getId)
                .process(
                        new KeyedProcessFunction<String, WaterSensor, String>() {

                            //1.声明状态
                            private AggregatingState<Integer,Double> avgVcState;

                            //2.初始化状态
                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                AggregatingStateDescriptor<Integer, Tuple2<Integer, Integer>, Double> avgVcStateDesc = new AggregatingStateDescriptor<>(
                                        "avgVcStateDesc",
                                        new AggregateFunction<Integer, Tuple2<Integer, Integer>, Double>() {

                                            //1.初始化累加器
                                            @Override
                                            public Tuple2<Integer, Integer> createAccumulator() {
                                                return Tuple2.of(0, 0);
                                            }

                                            //2.计算累计器中的值
                                            @Override
                                            public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> accumulator) {//accumulator:累加器
                                                return Tuple2.of(accumulator.f0 + value, accumulator.f1 + 1);
                                            }

                                            @Override
                                            public Double getResult(Tuple2<Integer, Integer> accumulator) {
                                                return (double) accumulator.f0 / accumulator.f1;
                                            }

                                            @Override
                                            public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> integerIntegerTuple2, Tuple2<Integer, Integer> acc1) {
                                                return null;
                                            }
                                        },
                                        Types.TUPLE(Types.INT, Types.INT)

                                );
                                //将状态描述传过去
                                avgVcState = runtimeContext.getAggregatingState(avgVcStateDesc);

                            }


                            @Override
                            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> out) throws Exception {
                            //3.使用状态
                                //将当前的水为值添加到状态中
                                avgVcState.add(value.getVc());

                                //输出
                                out.collect("传感器：" + context.getCurrentKey() + ", 平均水位为：" + avgVcState.get());


                            }
                        }
                )
                .print("PROCESS");




        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
